package com.zch.client.spring.mqtt.callback;

import com.alibaba.fastjson.JSON;
import com.zch.client.spring.mqtt.dto.MqttConsumerMessage;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
@Slf4j
public abstract class MqttClientRootCallback implements MqttCallback {
    @Override
    public void connectionLost(Throwable throwable) {
        onClosed(throwable);
    }
    public abstract void onClosed(Throwable throwable);
    //客户端收到消息
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        MqttConsumerMessage consumerMessage = new MqttConsumerMessage();
        consumerMessage.setMqttMessage(mqttMessage);
        consumerMessage.setTopic(topic);
        try{
            onMessage(consumerMessage);
        }catch (Throwable e){
            log.error("onMessage 异常:{}",e.getMessage());

        }
    }
    public abstract void onMessage(MqttConsumerMessage consumerMessage);
    /**
     * 消息已经成功发送到mqtt 服务器
     * @param iMqttDeliveryToken
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        IMqttAsyncClient asyncClient =  iMqttDeliveryToken.getClient();
        MqttMessage mqttMessage;
        try {
            mqttMessage = iMqttDeliveryToken.getMessage();
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
        String[] topics =  iMqttDeliveryToken.getTopics();
        String clientId = iMqttDeliveryToken.getClient().getClientId();
        String url = iMqttDeliveryToken.getClient().getServerURI();
        log.info("deliveryComplete clientId:{} 发布消息成功 topics:{},message:{}",clientId,topics, JSON.toJSONString(mqttMessage));
        try{
            onDeliverServer(mqttMessage);
        }catch (Throwable e){
            log.error("onDeliverServer 异常：{}",e.getMessage());
        }
    }
    // 单次消息送达服务端
    public abstract void onDeliverServer(MqttMessage mqttMessage);
}
